"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 二级分区
Case Name   : range_list二级分区表:锁--表级锁类型(行共享(row share))for update
Create At   : 2022/2/22
Owner       : @daiguatutu
Description :
    1:修改参数值lockwait_timeout=5s,partition_lock_upgrade_timeout=5,update_lockwait_timeout=5s
    2:创建表和表空间，并插入数据
    3:事务1-开启事务查询数据,等事务2/3执行结束提交
    4:事务2-对事务一的同张表进行共享语句操作
    5:事务3-对事务一的同张表进行冲突/不支持语句操作
    6.清理环境
Expect      :
    1:修改成功
    2:成功
    3:成功
    4:成功
    5:合理报错
    6.清理成功
History     :
    Modified by @daiguatutu 2022/8/3: 优化用例,增加等待时间,规避sql执行时间长导致执行顺序不一致的问题
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread
from testcase.utils.Logger import Logger


class Subpartition(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.t_name = 't_subpartition_0254'
        self.ts_name = 'ts_subpartition_0254'
        self.default_value1 = self.common.show_param('lockwait_timeout')
        self.default_value2 = self.common.show_param(
            'partition_lock_upgrade_timeout')
        self.default_value3 = self.common.show_param(
            'update_lockwait_timeout')

    def test_subpartition(self):
        step1_text = '-----step1:修改以下参数值;expect: 修改成功-----'
        self.log.info(step1_text)
        param_value = ['lockwait_timeout=5s',
                       'partition_lock_upgrade_timeout=5',
                       'update_lockwait_timeout=5s']
        for value in param_value:
            msg = self.pri_sh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{value}')
            self.assertTrue(msg, '执行失败:' + step1_text)

        step2_text = '-----step2:创建表和表空间，并插入数据;expect:成功-----'
        create_sql = f"drop table if exists {self.t_name};" \
            f"drop tablespace if exists {self.ts_name};" \
            f"create tablespace {self.ts_name} relative location " \
            f"'{self.ts_name}';" \
            f"create table if not exists {self.t_name}(" \
            f"col_1 int ,col_2 int ,col_3 varchar2 ( 30 ) ," \
            f"col_4 int)tablespace {self.ts_name} " \
            f"partition by range (col_1) subpartition by list (col_2) " \
            f"( partition p_range_1 values less than(-10) " \
            f"  ( subpartition p_list_1_1 values('-1','-2','-3','-4','-5')," \
            f"    subpartition p_list_1_2 values(default)), " \
            f"  partition p_range_2 values less than(20) " \
            f"  ( subpartition p_list_2_1 values('6','7','8','9','10'), " \
            f"    subpartition p_list_2_2 values(default)), " \
            f"  partition p_range_3 values less than(30) " \
            f"  ( subpartition p_list_3_1 values(default)), " \
            f"  partition p_range_4 values less than(40) " \
            f"  ( subpartition p_list_4_1 values(default)), " \
            f"  partition p_range_5 values less than(maxvalue) " \
            f") enable row movement; " \
            f"insert into {self.t_name} values (1,11,1,1),(4,41,4,4)," \
            f"(5,54,5,5);" \
            f"insert into {self.t_name} values(1,11,1,1),(4,41,4,4)," \
            f"(5,54,5,5),(8,87,8,8),(9,19,9,9)"
        msg = self.pri_sh.execut_db_sql(create_sql)
        self.log.info(msg)
        self.assertIn(self.constant.TABLESPCE_CREATE_SUCCESS, msg,
                      '执行失败:' + step2_text)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, msg,
                      '执行失败:' + step2_text)
        self.assertEqual(2, msg.count(self.constant.INSERT_SUCCESS_MSG),
                         '执行失败:' + step2_text)

        threads = []
        step3_text = '-----step3:事务1-开启事务查询数据,等事务2/3执行结束提交 expect:成功-----'
        self.log.info(step3_text)
        session1_sql = f'begin;' \
            f'select * from {self.t_name} for update; ' \
            f'SELECT PG_SLEEP(95);' \
            f'commit;'
        session1 = ComThread(self.pri_sh.execut_db_sql, args=(session1_sql,))
        threads.append(session1)

        step4_text = '-----step4:事务2-对事务一的同一张表进行共享操作 expect:成功-----'
        self.log.info(step4_text)
        session2_sql = f'SELECT PG_SLEEP(5);' \
            f'select * from {self.t_name};' \
            f'insert into {self.t_name} values(1,11,1,1),' \
            f'(4,41,4,4),(5,54,5,5);' \
            f'vacuum {self.t_name};' \
            f'analyze {self.t_name};' \
            f'create index  on {self.t_name}(col_1,col_2);' \
            f'reindex index {self.t_name}_col_1_col_2_tableoid_idx;'
        session2 = ComThread(self.pri_sh.execut_db_sql, args=(session2_sql,))
        threads.append(session2)

        step5_text = '-----step5:事务3-对事务一的同一张表进行表冲突/不支持操作 expect:合理报错-----'
        self.log.info(step5_text)
        session3_sql = f'SELECT PG_SLEEP(20);' \
            f'select * from {self.t_name} for update;' \
            f'select * from {self.t_name} for share;' \
            f'update {self.t_name} set col_4=80 where col_1=4;' \
            f'delete {self.t_name} where col_1 =4;' \
            f'drop table {self.t_name};' \
            f'truncate {self.t_name};' \
            f'create index  concurrently on {self.t_name}(col_1,col_2);' \
            f'cluster {self.t_name} using ' \
            f'{self.t_name}_col_1_col_2_tableoid_idx;' \
            f'vacuum full {self.t_name};'
        session3 = ComThread(self.pri_sh.execut_db_sql, args=(session3_sql,))
        threads.append(session3)

        self.log.info('-----启动线程-----')
        thread_results = []
        for t3 in threads:
            t3.setDaemon(True)
            t3.start()
        for t3 in threads:
            t3.join(100)
            thread_results.append(t3.get_result())
            self.log.info(t3.get_result())
        for error_msg in self.constant.SQL_WRONG_MSG:
            self.assertNotIn(error_msg, thread_results[1],
                             '执行失败:' + step4_text)
        self.assertEqual(8, thread_results[2].count(' Lock wait timeout'),
                         '执行失败:' + step5_text)
        self.assertIn('cannot create concurrent partitioned indexes',
                      thread_results[2], '执行失败:' + step5_text)

    def tearDown(self):
        self.log.info('-----step6:清理环境-----')
        text_1 = '-----删除表 expect:成功-----'
        self.log.info(text_1)
        del_cmd = f'drop table if exists {self.t_name};' \
            f'drop tablespace if exists {self.ts_name};'
        msg = self.pri_sh.execut_db_sql(del_cmd)
        self.log.info(msg)

        text_2 = '-----恢复参数默认值 expect:成功-----'
        self.log.info(text_2)
        res_cmd1 = self.pri_sh.execute_gsguc('reload',
                                             self.constant.GSGUC_SUCCESS_MSG,
                                             f'lockwait_timeout='
                                             f'{self.default_value1}')
        res_cmd2 = self.pri_sh.execute_gsguc('reload',
                                             self.constant.GSGUC_SUCCESS_MSG,
                                             f'partition_lock_upgrade_timeout='
                                             f'{self.default_value2}')
        res_cmd3 = self.pri_sh.execute_gsguc('reload',
                                             self.constant.GSGUC_SUCCESS_MSG,
                                             f'update_lockwait_timeout='
                                             f'{self.default_value3}')

        self.log.info('-----断言tearDown执行成功-----')
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, msg,
                      '执行失败:' + text_1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, msg, '执行失败:' + text_1)
        self.assertTrue(res_cmd1, '执行失败:' + text_2)
        self.assertTrue(res_cmd2, '执行失败:' + text_2)
        self.assertTrue(res_cmd3, '执行失败:' + text_2)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
